package com.cloud.job.consumer;

import com.cloud.api.domain.job.VehicleLogout;
import com.cloud.job.doris.DorisStreamLoad;
import com.cloud.job.utils.DateUtils;
import com.cloud.job.utils.JsonUtils;
import com.fasterxml.jackson.databind.node.ObjectNode;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.List;

/**
 * @author 杨斌
 * 2024-12-2024-12-18
 */
@Slf4j
@Component
@ConfigurationProperties
public class MessageLogoutConsumer {
    @Resource
    private DorisStreamLoad dorisStreamLoad;

    @KafkaListener(id="${listener.consumer3.id}",topics = "#{'${listener.consumer3.topics}'.split(',')}", groupId = "${listener.consumer3.groupId}",concurrency = "${listener.consumer3.concurrency}")
    public void processLoginMessage(@Payload List<ConsumerRecord<String, String>> consumerRecords,
                                    Acknowledgment acknowledgment) {
        log.info("Starting to process logout messages...");
        List<VehicleLogout> vehicleLogoutList = new ArrayList<>();
        try {

            log.info("Polled {} logout messages ",consumerRecords.size());
            if (consumerRecords.isEmpty()) {
                log.info("No logout messages to process");
                return;
            }
            for (ConsumerRecord<String, String> record : consumerRecords) {
                try {
                    VehicleLogout vehicleLogout = parseMessage(record.value());
                    vehicleLogoutList.add(vehicleLogout);
                } catch (Exception e) {
                    log.error("Error parsing logout messages: {}", record.value(), e);
                }
            }
            if (vehicleLogoutList.isEmpty()) {
                log.info("No parsed logout messages to insert into database");
                return;
            }
            String jsonString = JsonUtils.writeValueAsString(vehicleLogoutList);
            long loadStartTime = System.currentTimeMillis();
            dorisStreamLoad.loadJson(jsonString, VehicleLogout.class);
            long loadEndTime = System.currentTimeMillis();
            log.info("Inserted logout messages {} records into database in {} ms", vehicleLogoutList.size(), (loadEndTime - loadStartTime));
            // 提交偏移量
            acknowledgment.acknowledge();
            log.info("Committed offsets for processed logout messages.");

        } catch (Exception e) {
            log.error("Error processing logout messages", e);
        } finally {
            log.info("Finished processing logout messages.");
        }
    }
    private VehicleLogout parseMessage(String message) throws Exception {
        ObjectNode objectNode = JsonUtils.readObjectNode(message);
        VehicleLogout.VehicleLogoutBuilder vehicleLogoutBuilder = VehicleLogout.builder();
        String logoutTime = JsonUtils.getTextValue(objectNode, "dataContent", "localDateTime");
        String dt = DateUtils.parseAndFormatToDt(logoutTime);
        vehicleLogoutBuilder
                .logoutTime(logoutTime)
                .dt(dt)
                .logoutSerialNumber(JsonUtils.getIntValue(objectNode, "dataContent", "vehicleLogout", "logoutSerialNumber"))
                .receiveTime(JsonUtils.getTextValue(objectNode, "receive_time"))
                .vin(JsonUtils.getTextValue(objectNode, "header", "vin"))
                .record(JsonUtils.getTextValue(objectNode, "record"))
                .vehicleType(JsonUtils.getTextValue(objectNode, "vehicle_type"));
        return vehicleLogoutBuilder.build();
    }
}
